package eventbus

import (
	"context"
	"errors"
	"fmt"
	"reflect"
	"sync"
	"time"

	"github.com/samber/do/v2"
	"github.com/segmentio/kafka-go"
	"github.com/spf13/viper"

	"gitee.com/monkeyPP/happy-mall/pkg/log"
)

// KafkaEventBus 基于kafka-go的事件总线
type KafkaEventBus struct {
	writer     *kafka.Writer                 // kafka 生产者
	reader     *kafka.Reader                 // kafka 消费者
	serializer Serializer                    // 序列化器
	handlers   map[reflect.Type]EventHandler // 事件类型->处理函数映射
	mu         sync.RWMutex                  // 保护handlers的并发安全锁
	closed     bool                          // 是否关闭
	retryCount int                           // 处理失败重试次数
}

// KafkaConfig 事件总线配置
type KafkaConfig struct {
	Brokers      []string           `yaml:"brokers" mapstructure:"brokers"`         // Kafka broker地址列表
	Topic        string             `yaml:"topic" mapstructure:"topic"`             // 关联的Kafka主题
	GroupID      string             `yaml:"group_id" mapstructure:"group_id"`       // 消费组ID
	RetryCount   int                `yaml:"retry_count" mapstructure:"retry_count"` // 处理失败重试次数（默认3次）
	Serializer   Serializer         // 序列化器（默认JSON）
	ReaderConfig kafka.ReaderConfig // 自定义消费者配置
}

func NewKafkaEventBus(i do.Injector) (EventBus, error) {
	var cfg KafkaConfig
	err := viper.UnmarshalKey("kafka", &cfg)
	log.Infow("load kafka config", "config", cfg)
	if err != nil {
		panic(err)
	}
	// 默认配置
	if cfg.RetryCount <= 0 {
		cfg.RetryCount = 3
	}
	if cfg.Serializer == nil {
		cfg.Serializer = &JSONSerializer{}
	}

	// 初始化Kafka生产者
	writer := &kafka.Writer{
		Addr:     kafka.TCP(cfg.Brokers...),
		Topic:    cfg.Topic,
		Balancer: &kafka.LeastBytes{}, // 默认分区策略：最少字节数
	}

	// 初始化Kafka消费者
	readerCfg := kafka.ReaderConfig{
		Brokers:     cfg.Brokers,
		Topic:       cfg.Topic,
		GroupID:     cfg.GroupID,
		MinBytes:    1024,                   // 最小读取字节数
		MaxBytes:    1048576,                // 最大读取字节数（1MB）
		MaxWait:     500 * time.Millisecond, // 最大等待时间
		StartOffset: kafka.FirstOffset,      // 首次消费从最早开始
	}
	log.Infow("read kafka config", "config.Brokers", readerCfg.Brokers)
	// 合并自定义消费者配置
	// if cfg.ReaderConfig.GroupID != "" {
	// 	readerCfg.GroupID = cfg.ReaderConfig.GroupID
	// }
	// if cfg.ReaderConfig.StartOffset != 0 {
	// 	readerCfg.StartOffset = cfg.ReaderConfig.StartOffset
	// }
	reader := kafka.NewReader(readerCfg)

	return &KafkaEventBus{
		writer:     writer,
		reader:     reader,
		serializer: cfg.Serializer,
		handlers:   make(map[reflect.Type]EventHandler),
		retryCount: cfg.RetryCount,
	}, nil
}

func (k *KafkaEventBus) Subscribe(eventType any, handler EventHandler) error {
	eventTypeReflect := reflect.TypeOf(eventType)
	if eventTypeReflect.Kind() == reflect.Ptr {
		eventTypeReflect = eventTypeReflect.Elem()
	}

	k.mu.Lock()
	defer k.mu.Unlock()
	k.handlers[eventTypeReflect] = handler
	return nil
}

func (k *KafkaEventBus) Publish(ctx context.Context, event any) error {
	if k.closed {
		return errors.New("eventbus: 已关闭")
	}

	// 序列化事件
	payload, err := k.serializer.Marshal(event)
	if err != nil {
		return fmt.Errorf("序列化失败: %w", err)
	}

	// 发送到 kafka
	eventType := reflect.TypeOf(event)
	if eventType.Kind() == reflect.Ptr {
		eventType = eventType.Elem()
	}
	msg := kafka.Message{
		Key:   []byte(eventType.String()), // 用事件类型作为Key
		Value: payload,
		Time:  time.Now(),
	}

	// 发送消息（带超时控制）
	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
	defer cancel()
	if err := k.writer.WriteMessages(ctx, msg); err != nil {
		return fmt.Errorf("发送到Kafka失败: %w", err)
	}
	return nil
}

func (k *KafkaEventBus) Start(ctx context.Context) error {
	if k.closed {
		return errors.New("eventbus: 已关闭")
	}

	for {
		select {
		case <-ctx.Done():
			return ctx.Err() // 上下文取消，退出消费
		default:
			// 读取消息
			msg, err := k.reader.ReadMessage(ctx)
			if err != nil {
				// 非致命错误（如临时网络问题），重试
				if ctx.Err() != nil {
					return ctx.Err()
				}
				fmt.Printf("读取消息失败: %v，将重试\n", err)
				time.Sleep(1 * time.Second)
				continue
			}

			// 解析事件类型（从Key中获取）
			eventTypeStr := string(msg.Key)
			eventType, err := k.getTypeByName(eventTypeStr)
			if err != nil {
				fmt.Printf("未知事件类型: %s，跳过消息\n", eventTypeStr)
				if err := k.markAsProcessed(ctx, msg); err != nil {
					fmt.Printf("标记消息为已处理失败: %v\n", err)
				}
				continue
			}

			// 创建事件实例
			event := reflect.New(eventType).Interface()
			if err := k.serializer.Unmarshal(msg.Value, event); err != nil {
				fmt.Printf("反序列化事件失败: %v，跳过消息\n", err)
				if err := k.markAsProcessed(ctx, msg); err != nil {
					fmt.Printf("标记消息为已处理失败: %v\n", err)
				}
				continue
			}

			// 执行处理函数（带重试）
			if err := k.executeWithRetry(ctx, event); err != nil {
				fmt.Printf("事件处理失败（已重试%d次）: %v\n", k.retryCount, err)
				// 可在这里将失败消息发送到死信队列
				continue
			}

			// 处理成功，提交偏移量
			if err := k.markAsProcessed(ctx, msg); err != nil {
				fmt.Printf("提交偏移量失败: %v\n", err)
			}
		}
	}
}

// 执行处理函数并重试
func (k *KafkaEventBus) executeWithRetry(ctx context.Context, event any) error {
	eventType := reflect.TypeOf(event).Elem() // 取实际类型（如*OrderEvent -> OrderEvent）
	var lastErr error

	k.mu.RLock()
	handler, exists := k.handlers[eventType]
	k.mu.RUnlock()

	if !exists {
		return fmt.Errorf("未找到事件类型 %s 的处理函数", eventType.String())
	}

	// 重试逻辑
	for i := 0; i < k.retryCount; i++ {
		if err := handler(ctx, event); err != nil {
			lastErr = err
			fmt.Printf("第%d次处理失败: %v，将重试\n", i+1, err)
			time.Sleep(time.Duration(i+1) * 500 * time.Millisecond) // 指数退避
			continue
		}
		return nil // 处理成功
	}
	return fmt.Errorf("达到最大重试次数: %w", lastErr)
}

// 标记消息为已处理（提交偏移量）
func (k *KafkaEventBus) markAsProcessed(ctx context.Context, msg kafka.Message) error {
	return k.reader.CommitMessages(ctx, msg)
}

// 根据类型名获取反射类型
func (k *KafkaEventBus) getTypeByName(typeName string) (reflect.Type, error) {
	k.mu.RLock()
	defer k.mu.RUnlock()

	for typ := range k.handlers {
		if typ.String() == typeName {
			return typ, nil
		}
	}
	return nil, fmt.Errorf("类型 %s 未注册", typeName)
}

func (k *KafkaEventBus) Close() error {
	k.mu.Lock()
	defer k.mu.Unlock()

	if k.closed {
		return nil
	}

	k.closed = true
	var err error
	if err1 := k.reader.Close(); err1 != nil {
		err = errors.Join(err, err1)
	}
	if err2 := k.writer.Close(); err2 != nil {
		err = errors.Join(err, err2)
	}
	return err
}
